package com.cloud.easy.im.handler.qos;

import com.cloud.easy.im.BaseServerApplication;
import com.cloud.easy.im.argeement.Agreement;
import com.cloud.easy.im.netty.IMObserver;
import com.cloud.easy.im.utils.BasicThreadFactory;
import com.cloud.easy.im.utils.LocalSendHelper;
import com.cloud.easy.im.utils.SystemClock;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.*;

/**
 * S2C(对于服务端来说是发送)模式中QoS数据包质量包证机制之发送队列保证实现类.
 * <p>
 * 主流的移动端IM都有这个机制，具体流转如下：
 * 假设有A,B两个客户端发送数据并开启了Qos机制，使用了4个包：
 * A->S S->B B->S S->A（准确地说是两个包的两次转发，一个是IM消息包一个是IM应答包
 * <p>
 * note：
 * <p>
 * 本类和{@link QoS4RecivedDaemonRoot}是QoS机制的核心，目的是加强保证TCP协议在应用层的可靠性和送达率。
 * 当前QoS机制支持全部的C2C、C2S、S2C共3种消息交互场景下的消息送达质量保证：
 * <p>
 * 1)Client to Server(C2S)：即由某客户端主动发起，消息最终接收者是服务端，此模式下：重发由C保证、ACK应答由S发回；
 * 2)Server to Client(S2C)：即由服务端主动发起，消息最终接收者是某客户端，此模式下：重发由S保证、ACK应答由C发回；
 * 3)Client to Client(C2C)：即由客户端主动发起，消息最终接收者是另一客户端。此模式对于QoS机制来说，相当于C2S+S2C两程路径。
 *
 * @author wangjian
 * @version 0.1.0
 * @create 2020年10月10日3:06 下午
 */
public class QoS4SendDaemonRoot {

    private static Logger logger = LoggerFactory.getLogger(QoS4SendDaemonRoot.class);

    private BaseServerApplication serverApplication = null;
    /**
     * ConcurrentMap<FingerPrint, {@link Agreement}>
     */
    private ConcurrentMap<String, Agreement> sentMessages = new ConcurrentHashMap<String, Agreement>();
    /**
     * ConcurrentMap<FingerPrint, TimeStamp>
     */
    private ConcurrentMap<String, Long> sendMessagesTimestamp = new ConcurrentHashMap<String, Long>();
    /**
     * 执行间隔时间(S)
     */
    private int CHECH_INTERVAL = 5;
    /**
     * 每条数据的最小qos时间
     */
    private int MIN_MESSAGE_QOA_TIME = 2 * 1000;
    /**
     * qos最大尝试次数
     */
    private int QOS_TRY_COUNT = 2;
    private boolean excuting = false;
    /**
     * 本定时的作用是当worker启动或运行过程中出错时，可以自动进行恢复，而不至于丧失功能
     **/
    private final ScheduledExecutorService sendDaemon = new ScheduledThreadPoolExecutor(1,
            new BasicThreadFactory.Builder().namingPattern("qos-send-daemon-pool-%d").daemon(true).build());
    /**
     * TAG for log
     */
    private String TAG = "";

    public QoS4SendDaemonRoot(BaseServerApplication serverApplication) {
        this.serverApplication = serverApplication;
    }

    public QoS4SendDaemonRoot(int chechInterval
            , int minMessageQosTime
            , int qosTryCount
            , String tag) {
        if (chechInterval > 0) {
            this.CHECH_INTERVAL = chechInterval;
        }
        if (minMessageQosTime > 0) {
            this.MIN_MESSAGE_QOA_TIME = minMessageQosTime;
        }
        if (qosTryCount >= 0) {
            this.QOS_TRY_COUNT = qosTryCount;
        }
        this.TAG = tag;
    }

    private void doTaskOnece() {
        if (excuting) {
            return;
        }
        excuting = true;

        ArrayList<Agreement> lostMessages = new ArrayList<>();
        try {
            logger.debug("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】=========== 消息发送质量保证线程运行中, 当前需要处理的列表长度为" + sentMessages.size() + "...");

            for (Map.Entry<String, Agreement> entry : sentMessages.entrySet()) {
                String key = entry.getKey();
                final Agreement agreement = entry.getValue();
                if (agreement == null || !agreement.isQoS()) {
                    remove(key);
                    continue;
                }

                if (agreement.getRetryCount() >= QOS_TRY_COUNT) {
                    logger.debug("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】指纹为" + agreement.getFp()
                            + "的消息包重传次数已达" + agreement.getRetryCount() + "(最多" + QOS_TRY_COUNT + "次)上限，将判定为丢包！");


                    lostMessages.add((Agreement) agreement.clone());
                    remove(agreement.getFp());
                    continue;
                }

                long delta = SystemClock.getInstance().getNow() - sendMessagesTimestamp.get(key);
                if (delta <= MIN_MESSAGE_QOA_TIME) {
                    logger.warn("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】指纹为" + key + "的包距\"刚刚\"发出才" + delta
                            + "ms(<=" + MIN_MESSAGE_QOA_TIME + "ms将被认定是\"刚刚\"), 本次不需要重传哦.");
                    continue;
                }

                IMObserver sendResultObserver = (sendOK, extraObj) -> {
                    if (sendOK) {
                        logger.debug("【QoS4SendDaemonRoot" + TAG + "】【QoS发送方】指纹为" + agreement.getFp()
                                + "的消息包已成功进行重传，此次之后重传次数已达"
                                + agreement.getRetryCount() + "(最多" + QOS_TRY_COUNT + "次).");
                    } else {
                        logger.warn("【QoS4SendDaemonRoot" + TAG + "】【QoS发送方】指纹为" + agreement.getFp()
                                + "的消息包重传失败，它的重传次数之前已累计为"
                                + agreement.getRetryCount() + "(最多" + QOS_TRY_COUNT + "次).");

                    }
                };

                LocalSendHelper.sendData(agreement, sendResultObserver);
                agreement.increaseRetryCount();
            }
        } catch (Exception exception) {
            logger.warn("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】消息发送质量保证线程运行时发生异常," + exception.getMessage(), exception);
        }

        if (CollectionUtils.isEmpty(lostMessages)) {
            notifyMessageLost(lostMessages);
        }

        excuting = false;
    }

    /**
     * 通知消息被清理
     *
     * @param lostMessages
     */
    protected void notifyMessageLost(ArrayList<Agreement> lostMessages) {
        if (serverApplication != null && serverApplication.getMessageQoSEventListenerS2C() != null) {
            serverApplication.getMessageQoSEventListenerS2C().messagesLost(lostMessages);
        }
    }

    public QoS4SendDaemonRoot startup(boolean immediately) {
        stop();
        sendDaemon.scheduleAtFixedRate(this::doTaskOnece
                , immediately ? 0 : CHECH_INTERVAL
                , CHECH_INTERVAL, TimeUnit.SECONDS);

        logger.debug("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】=========== 消息发送质量保证线程已成功启动");
        return this;
    }

    public void stop() {
        if (sendDaemon.isShutdown()) {
            sendDaemon.shutdown();
        }
    }

    public boolean put(Agreement agreement) {
        if (agreement == null) {
            logger.warn(this.TAG + "Invalid arg p==null.");
            return false;
        }
        if (agreement.getFp() == null) {
            logger.warn(this.TAG + "Invalid arg p.getFp() == null.");
            return false;
        }

        if (!agreement.isQoS()) {
            logger.warn(this.TAG + "This protocal is not QoS pkg, ignore it!");
            return false;
        }

        if (hasRecieved(agreement.getFp())) {
            logger.warn("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】指纹为" + agreement.getFp()
                    + "的消息已经存在于接收列表中，该消息重复了（原理可能是对方因未收到应答包而错误重传导致），更新收到时间戳哦.");
        }

        sentMessages.put(agreement.getFp(), agreement);
        sendMessagesTimestamp.put(agreement.getFp(), SystemClock.getInstance().getNow());
        return true;
    }

    public boolean hasRecieved(String fingerPrintOfProtocal) {
        return sentMessages.containsKey(fingerPrintOfProtocal);
    }

    public void remove(final String fingerPrint) {
        try {
            sendMessagesTimestamp.remove(fingerPrint);
            Agreement result = sentMessages.remove(fingerPrint);
            logger.warn("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】指纹为" + fingerPrint + "的消息已成功从发送质量保证队列中移除(可能是收到接收方的应答也可能是达到了重传的次数上限)，重试次数="
                    + (result != null ? result.getRetryCount() : "none呵呵."));
        } catch (Exception e) {
            logger.warn("【QoS4SendDaemonRoot" + this.TAG + "】【QoS发送方】remove(fingerPrint)时出错了：", e);
        }
    }

    public void setServerApplication(BaseServerApplication serverApplication) {
        this.serverApplication = serverApplication;
    }
}
